package org.zjt.spark.dstream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  *
  * linux producer ：nc -lk 9002
  *
  * 1、transform、join（内连接）、leftoutjoin 外连接   out.... 都是外连接
  */
object TransFormTest extends App {

  val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount").set("spark.driver.allowMultipleContexts","true")
  val ssc = new StreamingContext(conf, Seconds(10))
  val blacklist = ssc.sparkContext.parallelize(Array("zhang")).map((_,false))
  val resource = ssc.socketTextStream("centos", 9002)


  val meg =resource.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)

  // TODO: 过滤黑名单
  meg.transform( rdd => rdd.leftOuterJoin(blacklist) ).filter( a =>{a._2._2.isEmpty || a._2._2.get}).print()

  ssc.start()
  ssc.awaitTermination()
  ssc.stop()

}
